package mqtt

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"time"

	mc "gitee.com/kitfast/letsgo/config/mq/mqtt"
	"gitee.com/kitfast/letsgo/io/file"
	"gitee.com/kitfast/letsgo/io/log"
	mq "github.com/eclipse/paho.mqtt.golang"
)

const (
	AT_MOST_ONE  = 0 // 最多一次
	AT_LEAST_ONE = 1 // 至少一次
	EXACTLY_ONE  = 2 // 只有一次
)

// 消息
type Message mq.Message

// 消息处理
type MessageHandler func(Message)

type sub struct {
	topic   string
	qos     byte
	handler mq.MessageHandler
}

// MQTT 客户端
type Mqtt struct {
	conf   *mc.MqttConfig
	logger log.Logger

	opts   *mq.ClientOptions
	client mq.Client

	closed chan struct{}

	subs map[string]*sub
}

func NewMqtt(c *mc.MqttConfig, logger log.Logger) *Mqtt {
	lg := logger
	if g, ok := logger.(*log.WrappedLogger); ok {
		lg = g.WithPrefix(func() string { return "[MQTT]" })
	}
	m := &Mqtt{conf: c, logger: lg, closed: make(chan struct{}), subs: make(map[string]*sub)}
	m.init()
	return m
}

func (m *Mqtt) init() {
	m.opts = mq.NewClientOptions()
	m.opts.AddBroker(m.conf.Broker)
	m.opts.SetClientID(m.conf.Client)
	m.opts.SetUsername(m.conf.Username)
	m.opts.SetPassword(m.conf.Password)
	if m.conf.Tls != nil {
		m.initTls()
	}
	m.opts.SetConnectRetry(true)
	m.opts.SetDefaultPublishHandler(func(c mq.Client, msg mq.Message) {
		m.logger.Info("Received message: %s [%d] %s", msg.Topic(), msg.MessageID(), string(msg.Payload()))
		msg.Ack()
	})
	m.opts.OnConnect = func(c mq.Client) {
		for _, s := range m.subs {
			m.subscribe(s)
		}
		m.logger.Info("Connected: %s", m.conf.Broker)
	}
	m.opts.OnConnectionLost = func(c mq.Client, err error) {
		m.logger.Info("Connection lost: %s, error: %v", m.conf.Broker, err)
	}
	m.client = mq.NewClient(m.opts)
}

func (m *Mqtt) initTls() {
	t := m.conf.Tls
	c := &tls.Config{}
	if t.Ca != "" {
		cp := x509.NewCertPool()
		ca, err := file.ReadToString(t.Ca)
		if err != nil {
			m.logger.Error("load ca error: %v", err)
			return
		}
		cp.AppendCertsFromPEM([]byte(ca))
		c.RootCAs = cp
	}
	if t.Cert != "" && t.Key != "" {
		kp, err := tls.LoadX509KeyPair(t.Cert, t.Key)
		if err != nil {
			m.logger.Error("load client certifacate error: %v", err)
			return
		}
		c.ClientAuth = tls.NoClientCert
		c.ClientCAs = nil
		c.Certificates = []tls.Certificate{kp}
	}
	m.opts.SetTLSConfig(c)
}

func (m *Mqtt) Connect() {
	token := m.client.Connect()
	if !token.WaitTimeout(time.Minute) {
		m.logger.Error("Conntect to %s timeout (60s)", m.conf.Broker)
		return
	}
	if err := token.Error(); err != nil {
		m.logger.Error("Conntect to %s fail, error: %v", m.conf.Broker, err)
		return
	}
}

func (m *Mqtt) Close() {
	m.client.Disconnect(30000)
	close(m.closed)
}

func (m *Mqtt) PublishJson(topic string, qos byte, body any) {
	j, err := json.Marshal(body)
	if err != nil {
		m.logger.Error("Publish to %s@%s fail, error: %v", topic, m.conf.Broker, err)
		return
	}
	token := m.client.Publish(topic, qos, false, j)
	if !token.WaitTimeout(15 * time.Second) {
		m.logger.Error("Publish to %s@%s timeout (15s)", topic, m.conf.Broker)
		return
	}
	if err := token.Error(); err != nil {
		m.logger.Error("Publish to %s@%s fail, error: %v", topic, m.conf.Broker, err)
		return
	}
}

func (m *Mqtt) subscribe(s *sub) {
	token := m.client.Subscribe(s.topic, s.qos, s.handler)
	if !token.WaitTimeout(15 * time.Second) {
		m.logger.Error("Subscribe from %s@%s timeout (15s)", s.topic, m.conf.Broker)
		return
	}
	if err := token.Error(); err != nil {
		m.logger.Error("Subscribe from %s@%s fail, error: %v", s.topic, m.conf.Broker, err)
		return
	}
}

func (m *Mqtt) Subscribe(topic string, qos byte, callback MessageHandler) {
	handler := func(c mq.Client, m mq.Message) { callback(m) }
	s := &sub{topic: topic, qos: qos, handler: handler}
	m.subs[topic] = s
	m.subscribe(s)
}

func (m *Mqtt) Listen() {
	<-m.closed
}
